//
//  WebSocket.m
//  Socket.IO-Client
//
//  Created by alanwang.dev on 16/8/8.
//  Copyright © 2016年 com.alanwang.dev. All rights reserved.
//

#import "WebSocket.h"
#import <Security/CipherSuite.h>
#import "NSData+SubData.h"

static NSString *ErrorDomain = @"WebSocket";

@interface WebSocket ()

@property (nonatomic, copy) NSArray *optionalProtocols;
@property (nonatomic, copy, readonly) NSString * headerWSUpgradeName;
@property (nonatomic, copy, readonly) NSString * headerWSUpgradeValue;
@property (nonatomic, copy, readonly) NSString * headerWSHostName;
@property (nonatomic, copy, readonly) NSString * headerWSConnectionName;
@property (nonatomic, copy, readonly) NSString * headerWSConnectionValue;
@property (nonatomic, copy, readonly) NSString * headerWSProtocolName;
@property (nonatomic, copy, readonly) NSString * headerWSVersionName;
@property (nonatomic, copy, readonly) NSString * headerWSVersionValue;
@property (nonatomic, copy, readonly) NSString * headerWSKeyName;
@property (nonatomic, copy, readonly) NSString * headerOriginName;
@property (nonatomic, copy, readonly) NSString * headerWSAcceptName;
@property (nonatomic, assign, readonly) NSInteger BUFFER_MAX;
@property (nonatomic, assign, readonly) uint8_t FinMask;
@property (nonatomic, assign, readonly) uint8_t OpCodeMask;
@property (nonatomic, assign, readonly) uint8_t RSVMask;
@property (nonatomic, assign, readonly) uint8_t MaskMask;
@property (nonatomic, assign, readonly) uint8_t PayloadLenMask;
@property (nonatomic, assign, readonly) NSInteger MaxFrameSize;
@property (nonatomic, strong) NSURL *currentURL;
@property (nonatomic, strong) NSURL *url;
@property (atomic, strong) NSInputStream *inputStream;
@property (atomic, strong) NSOutputStream *outputStream;
@property (atomic, strong) NSOperationQueue *writeQueue;
@property (atomic, strong) NSMutableArray<WSResponse *> *readStack;
@property (atomic, strong) NSMutableArray *inputQueue;
@property (atomic, assign) BOOL connected;
@property (atomic, assign) BOOL isCreated;
@property (atomic, assign) BOOL certValidated;
@property (atomic, assign) BOOL didDisconnect;
@property (atomic, assign) BOOL readyToWrite;
@property (atomic, strong) NSData *fragBuffer;
//@property (atomic, assign) BOOL canDispatch;
@property (atomic, strong) NSLock *mutex;
@property (atomic, strong) dispatch_queue_t sharedWorkQueue ;
@end

@implementation WebSocket


- (id) initWithURL:(NSURL *)url protocols:(NSArray<NSString *> *)protocols {
  if (self = [super init]) {
    _queue = dispatch_get_main_queue();
    _url = url;
    _origin = url.absoluteString;
    _writeQueue = [[NSOperationQueue alloc] init];
    _writeQueue.maxConcurrentOperationCount = 1;
    _optionalProtocols = protocols;
    
    _headerWSUpgradeName = @"Upgrade";
    _headerWSUpgradeValue = @"websocket";
    _headerWSHostName = @"Host";
    _headerWSConnectionName = @"Connection";
    _headerWSConnectionValue = @"Upgrade";
    _headerWSProtocolName = @"Sec-WebSocket-Protocol";
    _headerWSVersionName = @"Sec-WebSocket-Version";
    _headerWSVersionValue = @"13";
    _headerWSKeyName = @"Sec-WebSocket-Key";
    _headerOriginName = @"Origin";
    _headerWSAcceptName = @"Sec-WebSocket-Accept";
    _BUFFER_MAX = 4096;
    _FinMask = 0x80;
    _OpCodeMask = 0x0f;
    _RSVMask = 0x70;
    _MaskMask = 0x80;
    _PayloadLenMask = 0x7f;
    _MaxFrameSize = 32;
    _timeout= 5;
    _inputQueue = [NSMutableArray array];
    _readStack = [NSMutableArray array];
    _sharedWorkQueue  = dispatch_queue_create("com.vluxe.starscream.websocket", DISPATCH_QUEUE_SERIAL);
    _mutex = [[NSLock alloc] init];
  }
  
  return self;
}

- (BOOL) canDispatch {
  [self.mutex lock];
  BOOL canWork = _readyToWrite;
  [self.mutex unlock];
  return canWork;
}

- (BOOL) isConnected {
  return _connected;
}
- (NSMutableDictionary *) headers {
  return [NSMutableDictionary dictionaryWithCapacity:0];
}

- (void) wirteString:(NSString *)str completion:(CompletionBlock)completion {
  if (!self.isConnected) {
    return;
  }
  [self dequeueWrite:[str dataUsingEncoding:NSUTF8StringEncoding].mutableCopy code:TextFrame writeCompletion:completion];
}

- (void) writeData:(NSData *)data completion:(CompletionBlock)completion {
  if (!self.isConnected) {
    return;
  }
  [self dequeueWrite:data.mutableCopy code:BinaryFrame writeCompletion:completion];
}

- (void) writePing:(NSData *)data completion:(CompletionBlock)completion {
  if (!self.isConnected) {
    return;
  }
  [self dequeueWrite:data.mutableCopy code:Ping writeCompletion:completion];
}

- (void) createHTTPRequest {
  CFStringRef requestMethod = CFSTR("GET");
  CFHTTPMessageRef urlRequest = CFHTTPMessageCreateRequest(kCFAllocatorDefault, requestMethod, (__bridge CFURLRef _Nonnull)(self.url), kCFHTTPVersion1_1);
  NSNumber *port = self.url.port;
  if (port == nil) {
    if ([@[@"wss", @"https"] containsObject:self.url.scheme]) {
      port = @433;
    } else {
      port = @80;
    }
  }
  
  [self addHeader:urlRequest key:self.headerWSUpgradeName value:self.headerWSUpgradeValue];
  [self addHeader:urlRequest key:self.headerWSConnectionName value:self.headerWSConnectionValue];
  if (self.optionalProtocols) {
    [self addHeader:urlRequest key:self.headerWSProtocolName value:[self.optionalProtocols componentsJoinedByString:@","]];
  }
  [self addHeader:urlRequest key:self.headerWSVersionName value:self.headerWSVersionValue];
  [self addHeader:urlRequest key:self.headerWSKeyName value:[self generateWebSocketKey]];
  if (self.origin) {
    [self addHeader:urlRequest key:self.headerOriginName value:self.origin];
  }
  [self addHeader:urlRequest key:self.headerWSHostName value:[NSString stringWithFormat:@"%@:%@",self.url.host,port]];
  
  [self.headers enumerateKeysAndObjectsUsingBlock:^(id  _Nonnull key, id  _Nonnull obj, BOOL * _Nonnull stop) {
    [self addHeader:urlRequest key:key value: obj];
  }];
  CFDataRef cfHTTPMessage = CFHTTPMessageCopySerializedMessage(urlRequest);
  [self initStreamWithData:(__bridge_transfer NSData *)cfHTTPMessage port:port.intValue];
}

- (void) addHeader:(CFHTTPMessageRef )urlRequset key:(NSString *)key value:(NSString *)value {
  NSLog(@"headerKey:%@, value:%@",key, value);
  CFHTTPMessageSetHeaderFieldValue(urlRequset, (__bridge CFStringRef)key, (__bridge CFStringRef)value);
}

// 生成ws key
- (NSString *) generateWebSocketKey {
  NSString *key = @"";
  NSInteger seed = 16;
  for (int i = 0; i< seed; i++) {
    char uni = 97 + arc4random_uniform(25);
    key = [key stringByAppendingFormat:@"%c",uni];
  }
  NSData *data = [key dataUsingEncoding:NSUTF8StringEncoding];
  NSData *baseKey = [data base64EncodedDataWithOptions:NSDataBase64Encoding64CharacterLineLength];
  return [NSString stringWithUTF8String:[baseKey bytes]];
}


- (void) initStreamWithData:(NSData *)data port:(int)port {
  CFReadStreamRef readStream;
  CFWriteStreamRef writeStream;
  
  NSString *h = self.url.host;
  CFStreamCreatePairWithSocketToHost(NULL,(__bridge CFStringRef)h, port, &readStream, &writeStream);
  self.inputStream = (__bridge_transfer  NSInputStream *)(readStream);
  self.outputStream = (__bridge_transfer  NSOutputStream *)(writeStream);
  self.inputStream.delegate = self;
  self.outputStream.delegate = self;
  if ([@[@"wss", @"https"] containsObject:self.url.scheme]) {
    [self.inputStream setProperty:NSStreamSocketSecurityLevelNegotiatedSSL forKey:NSStreamSocketSecurityLevelKey];
    [self.outputStream setProperty:NSStreamSocketSecurityLevelNegotiatedSSL forKey:NSStreamSocketSecurityLevelKey];
  } else {
    self.certValidated = NO;
  }
  if (self.voipEnabled) {
    [self.inputStream setProperty:NSStreamNetworkServiceTypeVoIP forKey:NSStreamNetworkServiceType];
    [self.outputStream setProperty:NSStreamNetworkServiceTypeVoIP forKey:NSStreamNetworkServiceType];
  }
  if (self.selfSignedSSL) {
    NSDictionary *sslpro = @{(__bridge NSString *)kCFStreamSSLValidatesCertificateChain:(__bridge NSNumber*)kCFBooleanFalse,
                             (__bridge NSString *)kCFStreamSSLPeerName:(__bridge NSNull *)kCFNull};
    
    [self.inputStream setProperty:sslpro forKey:(__bridge NSString *)kCFStreamPropertySSLSettings];
    [self.inputStream setProperty:sslpro forKey:(__bridge NSString *)kCFStreamPropertySSLSettings];
  }
  if (self.enabledSSLCipherSuites) {
    SSLContextRef sslContextIn =  (SSLContextRef)CFReadStreamCopyProperty(readStream, kCFStreamPropertySSLContext);
    SSLContextRef sslContextOut = (SSLContextRef)CFWriteStreamCopyProperty(writeStream, kCFStreamPropertySSLContext);
    
    SSLCipherSuite *suite = malloc(sizeof(SSLCipherSuite)*self.enabledSSLCipherSuites.count);
    int i = 0;
    for (NSNumber *number in self.enabledSSLCipherSuites) {
      memset(&suite[i], [number unsignedShortValue], sizeof(SSLCipherSuite));
      i++;
    }
    OSStatus sslIn =  SSLSetEnabledCiphers(sslContextIn, suite,self.enabledSSLCipherSuites.count);
    OSStatus sslOut = SSLSetEnabledCiphers(sslContextOut, suite,self.enabledSSLCipherSuites.count);
    if (sslIn != errSecSuccess) {
      NSError *error = [self errorWithDetail:@"Error setting ingoing cypher suites" code:sslIn];
      [self disconnectStream: error];
      return;
    }
    if (sslOut != errSecSuccess) {
      NSError *error = [self errorWithDetail:@"Error setting outgoing cypher suites" code:sslIn];
      [self disconnectStream: error];
      return;
    }
  }
  CFReadStreamSetDispatchQueue(readStream, _sharedWorkQueue);
  CFWriteStreamSetDispatchQueue(writeStream, _sharedWorkQueue);
  CFReadStreamOpen(readStream);
  CFWriteStreamOpen(writeStream);
  [self.mutex lock];
  _readyToWrite = YES;
  [self.mutex unlock];
  __block NSInteger timeout = self.timeout * 10000000;
  [self.writeQueue addOperationWithBlock:^{
    while (![self.outputStream hasSpaceAvailable]) {
      usleep(100* 20);
      timeout-= 100* 20;
      if (timeout < 0) {
        [self cleanupStream];
        [self doDisconnect:[self errorWithDetail:@"write wait timed out" code:2]];
        return ;
      } else if ([self.outputStream streamError]){
        return;
      }
    }
    NSInteger ret = [self.outputStream write:[data bytes] maxLength:data.length];
    NSLog(@"outputstream ret:%ld",ret);
  }];
}


- (void) disconnectStream:(NSError *) error {
  if (error == nil) {
    [self.writeQueue waitUntilAllOperationsAreFinished];
  } else {
    [self.writeQueue cancelAllOperations];
  }
    [self cleanupStream];
    [self doDisconnect:error];
}


- (void) cleanupStream {
  _outputStream.delegate = nil;
  _inputStream.delegate = nil;
  if (_inputStream) {
    CFReadStreamSetDispatchQueue((__bridge CFReadStreamRef)_inputStream, nil);
    [_inputStream close];
  }
  if (_outputStream) {
    CFWriteStreamSetDispatchQueue((__bridge CFWriteStreamRef)_outputStream, nil);
    [_outputStream close];
  }
  
  _inputStream = nil;
  _outputStream = nil;
}

- (void) processInputStream {
  UInt8 *buff = malloc(_BUFFER_MAX);
  NSInteger length = [self.inputStream read: buff maxLength:_BUFFER_MAX];
  NSLog(@"&&&&&&:%@",@(length));
  if (length < 0) {
     free(buff);
    return;
  }
  BOOL process = NO;
  if (self.inputQueue.count == 0) {
    process = YES;
  }
  [self.inputQueue addObject:[NSData dataWithBytes:buff length:length]];
  if (process) {
    [self dequeueInput];
  }
  
  free(buff);
}

- (void) dequeueInput {
  while (self.inputQueue.count != 0) {
    NSData *data = _inputQueue[0];
    NSMutableData *combine = [NSMutableData dataWithData:_fragBuffer.copy];
    [combine appendData:data];
    self.fragBuffer = nil;
    
    if (!self.connected) {
      [self processTCPHandshake:combine bufferLen:combine.length];
    } else {
      [self processRawMessagesInBuffer:combine bufferLen:combine.length];
    }
    [self.inputQueue removeObjectAtIndex:0];
  }
}

- (void) processTCPHandshake:(NSData *) buffer bufferLen:(NSInteger)bufferLen {
  NSInteger code = [self processHTTP:buffer length:bufferLen];
  switch (code) {
    case 0: {
      _connected = YES;
      __weak typeof(self) weakSelf = self;
      if (self.canDispatch) {
        dispatch_async(_queue, ^{
          if (self.onConnect) {
            weakSelf.onConnect();
          }
          [weakSelf.deleate  websocketDidConnect:weakSelf];
        });
      }
    }
      break;
    case -1: {
      _fragBuffer = buffer;
    }
      break;
    default:
      [self doDisconnect: [self errorWithDetail:@"Invalid HTTP upgrade" code:code]];
      break;
  }
}


- (NSInteger )processHTTP:(NSData *)buffer length:(NSInteger)length {
  UInt8 *bytes = (UInt8 *)[buffer bytes];
  int crlf[4] = {'\r','\n','\r','\n'};
  int k = 0;
  NSInteger totalSize = 0;
  for (NSInteger i = 0 ; i < length; i++) {
    if (bytes[i] == crlf[k]) {
      k++;
      if (k == 3) {
        totalSize = i + 1;
        break;
      }
    } else {
      k = 0;
    }
  }
  if (totalSize > 0) {
    int code = [self validateResponse:buffer bufferLen:totalSize];
    if (code != 0) {
      return code;
    }
    totalSize = totalSize + 1;
    NSInteger resetSize = length - totalSize;
    if (resetSize > 0) {
      [self processRawMessagesInBuffer:[buffer dataFromIndex:totalSize] bufferLen:resetSize];
    }
    return 0;
  }
  return -1;
}

- (void) connect {
  if (self.isCreated) {
    return;
  }
  self.didDisconnect = NO;
  self.isCreated = YES;
  [self createHTTPRequest];
  self.isCreated = NO;
}

- (void) disconnect:(NSTimeInterval)forceTimeout {
  if (forceTimeout > 0) {
    dispatch_after(dispatch_time(DISPATCH_TIME_NOW, (int64_t)(forceTimeout * NSEC_PER_SEC)), dispatch_get_main_queue(), ^{
      [self disconnectStream:nil];
    });
  } else if (forceTimeout < 0) {
      [self writeError:Normal];
  } else {
    [self disconnectStream:nil];
  }
}


- (int) validateResponse:(NSData *)data bufferLen:(NSInteger)bufferLen {
  CFHTTPMessageRef response = CFHTTPMessageCreateEmpty(kCFAllocatorDefault, false);
  CFHTTPMessageAppendBytes(response, [data bytes], bufferLen);
  CFIndex  code = CFHTTPMessageGetResponseStatusCode(response);
  if (code != 101) {
      CFRelease(response);
    return (int)code;
  }
  CFDictionaryRef cfHeaders = CFHTTPMessageCopyAllHeaderFields(response);
  NSDictionary *dic = (__bridge_transfer NSDictionary*)cfHeaders;
  NSString *name = [dic objectForKey:self.headerWSAcceptName];
  if (name.length > 0) {
    CFRelease(response);
    return 0;
  }
  CFRelease(response);
  return -1;
}

- (void) processRawMessagesInBuffer:(NSData *)pointer bufferLen:(NSInteger)bufferLen {
//    return;
  while (pointer.length >= 2) {
    pointer = [self processOneRawMessage:pointer];
  }
  
  if (pointer.length > 0) {
    _fragBuffer = pointer;
  }
}

- (NSData *) processOneRawMessage:(NSData *)buffer {
  WSResponse *response =  self.readStack.lastObject;
  UInt8 baseAddress = [buffer dataAtIndex:0];
  unsigned long bufferLen = buffer.length;
  if (!response && bufferLen < 2) {
    self.fragBuffer = buffer;
    return [NSData data];
  }
  if (response.bytesLeft > 0) {
    NSInteger len = response.bytesLeft;
    NSInteger extra = bufferLen - response.bytesLeft;
    if (response.bytesLeft > bufferLen) {
      len = bufferLen;
      extra = 0;
    }
    response.bytesLeft -= len;
    char *a = malloc(len);
    memset(a, 0, len);
    [response.buffer appendData:[NSData dataWithBytes:a length:len]];
    [self processResponse:response];
      free(a);
    return nil;
  } else {
    UInt8 isFin = (_FinMask & baseAddress);
    OpCode receivedOpcode = (_OpCodeMask & baseAddress);
    UInt8 address2 = [buffer dataAtIndex:1];
    UInt8 isMasked = _MaskMask & address2;
    UInt8 payloadLen = _PayloadLenMask & address2;
    int offset = 2;
    if ((isMasked > 0 ||(_RSVMask & baseAddress)
         > 0 )&& receivedOpcode != Pong) {
      CloseCode errorCode = ProtocolError;
      [self doDisconnect:[self errorWithDetail:@"unknown opcode" code:errorCode]];
      [self writeError:errorCode];
      return nil;
    }
    BOOL isControlFrame = (receivedOpcode == ConnectionClose || receivedOpcode == Ping);
    if (!isControlFrame && (receivedOpcode != BinaryFrame && receivedOpcode != ContinueFrame
                            && receivedOpcode != TextFrame && receivedOpcode != Pong)) {
      CloseCode errorCode = ProtocolError;
      [self doDisconnect:[self errorWithDetail:@"unknown opcode" code:errorCode]];
      [self writeError:errorCode];
      return nil;
    }
    if (isControlFrame && isFin == 0) {
      CloseCode errorCode = ProtocolError;
      [self doDisconnect:[self errorWithDetail:@"control frames can't be fragmented" code:errorCode]];
      [self writeError:errorCode];
      return nil;
    }
    if (receivedOpcode == ConnectionClose) {
      CloseCode code = Normal;
      if (payloadLen == 1) {
        code = ProtocolError;
      } else if (payloadLen > 1) {
        code = [[self class] readUint16:buffer offset:offset];
        if (code < 1000 || (code > 1003 && code < 1007) || (code > 1011)) {
          code = ProtocolError;
        }
        offset += 2;
      }
      
      if (payloadLen > 2) {
        int len = (int)payloadLen-2;
        if (len > 0) {
          NSData *data = [buffer subdataWithRange:NSMakeRange(baseAddress+offset, len)];
          NSString *str = [[NSString alloc] initWithData:data encoding:NSUTF8StringEncoding];
          if (str == nil) {
            code = ProtocolError;
          }
        }
      }
      [self doDisconnect:[self errorWithDetail:@"connection closed by server" code:code]];
      [self writeError: code];
      return nil;
    }
    
    if (isControlFrame && payloadLen > 125) {
      [self writeError: ProtocolError];
    }
    UInt64 dataLength = (UInt64)(payloadLen);
    if (dataLength == 127) {
      dataLength = [[self class] readUint64:buffer offset:offset];
      offset += sizeof(UInt64);
    } else if (dataLength == 126) {
      dataLength = (UInt64)[[self class] readUint16:buffer offset:offset];
      offset += sizeof(UInt16);
    }
    if (bufferLen < offset || (bufferLen-offset) < dataLength) {
      _fragBuffer = [buffer subdataWithRange:NSMakeRange(0, bufferLen)];
    }
    UInt16 len = dataLength;
    if (dataLength > bufferLen) {
      len = bufferLen - offset;
    }
    NSData *data = nil;
    if (len < 0) {
      len = 0;
      data = [NSData data];
    } else {
      data = [buffer subdataWithRange:NSMakeRange(offset, len)];
    }
    if (receivedOpcode == Pong) {
      if (self.canDispatch) {
        __weak typeof(self) weakself = self;
        dispatch_async(_queue, ^{
            if (weakself.onPong) {
                weakself.onPong();
            }
          
          [weakself.pongDelegate websocketDidReceivePong:weakself];
        });
      }
      return [buffer dataFromIndex:offset+len];
    }
    WSResponse *response = _readStack.lastObject;
    if (isControlFrame) {
      response = nil;
    }
    if (isFin == 0 && receivedOpcode == ContinueFrame &&response == nil) {
      CloseCode errorCode = ProtocolError;
      [self doDisconnect:[self errorWithDetail:@"continue frame before a binary or text frame" code:errorCode]];
      [self writeError: errorCode];
      return nil;
    }
    BOOL isNew = NO;
    if (response == nil) {
      if (receivedOpcode == ContinueFrame) {
        CloseCode errCode = ProtocolError;
        [self doDisconnect:[self errorWithDetail:@"first frame can't be a continue frame" code:errCode]];
        [self writeError:errCode];
        return nil;
      }
      isNew = YES;
      response = [[WSResponse alloc] init];
      response.code = receivedOpcode;
      response.bytesLeft = (NSInteger)dataLength;
      response.buffer = [NSMutableData dataWithData:data];
    } else {
      if (receivedOpcode == ContinueFrame) {
        response.bytesLeft = (NSInteger)dataLength;
      } else {
        CloseCode errorCode = ProtocolError;
        [self doDisconnect:[self errorWithDetail:@"second and beyond of fragment message must be a continue frame" code:errorCode]];
        [self writeError: errorCode];
        return nil;
      }
      [response.buffer appendData:data];
    }
    if (response) {
      response.bytesLeft -= len;
      response.frameCount += 1;
      response.isFin = isFin > 0 ? YES : NO;
      if (isNew) {
        [_readStack addObject:response];
      }
      [self processResponse:response];
    }
    int step = offset + (int)len;
    return [buffer dataFromIndex:step];
  }
}


- (void)dequeueWrite:(NSMutableData *)data code:(OpCode)code writeCompletion:(void(^)(void)) writeCompletion {
  
  __weak typeof(self) weakSelf = self;
  [self.writeQueue addOperationWithBlock:^{
      __strong typeof(weakSelf) strongSelf=weakSelf;
    int offset = 2;
    Byte *bytes = [data bytes];
    NSInteger dataLength = data.length;
    NSMutableData *frame = [[NSMutableData alloc]initWithCapacity:dataLength+strongSelf.MaxFrameSize];
    Byte *buffer = frame.mutableBytes;
    buffer[0] = strongSelf.FinMask | code;
    if (dataLength < 126) {
      buffer[1] = (UInt8) dataLength;
    } else if (dataLength < 0xffff) {
      buffer[1] = 126;
      [[strongSelf class] writeUint16:frame offset:offset value:(UInt16)(dataLength&0xffff)];
      offset += sizeof(UInt16);
    } else {
      buffer[1] = 127;
      [[strongSelf class] writeUint64:frame offset:offset value:(UInt64)(dataLength&0xffff)];
      offset += sizeof(UInt64);
    }
    buffer[1] |= strongSelf.MaskMask;
    //????
    Byte *maskKey = &buffer[offset];
    SecRandomCopyBytes(kSecRandomDefault, sizeof(UInt32), maskKey);
    offset += sizeof(UInt32);
    for (NSInteger i = 0; i < dataLength; i++) {
      buffer[offset] = bytes[i] ^ maskKey[i % sizeof(UInt32)];
      offset += 1;
    }
    NSInteger totoal = 0;
    while (YES) {
      NSLog(@"write ----------");
      NSInteger len = [_outputStream write:&buffer[totoal] maxLength:offset-totoal];
      NSLog(@"len:%@",@(len));
      if (len < 0) {
        NSError *error;
        if (_outputStream.streamError) {
          error = _outputStream.streamError;
        } else {
          error = [strongSelf errorWithDetail:@"output stream error during write" code:OutputStreamWriteError];
        }
        [strongSelf doDisconnect:error];
        break;
      } else {
        totoal += len;
      }
      if (totoal >= offset) {
        if (writeCompletion) {
          dispatch_async(_queue, writeCompletion);
        }
        break;
      }
    }
  }];
}

- (void) doDisconnect:(NSError *)error {
  if (_didDisconnect) {
    return;
  }
  _didDisconnect = YES;
  _connected = NO;
  if (!self.canDispatch) {
    return;
  }
  __weak typeof(self) weakSelf = self;
  dispatch_async(_queue, ^{
    if (weakSelf.onDisconnect) {
      weakSelf.onDisconnect(error);
    }
    [weakSelf.deleate websocketDidDisconnect:weakSelf error:error];
  });
  
}

- (void) writeError:(NSUInteger)error {
  
  NSMutableData *buf = [NSMutableData dataWithCapacity:sizeof(UInt16)];
  [[self class] writeUint16:buf offset:0 value:error];
  [self dequeueWrite:buf code:ConnectionClose writeCompletion:^{
  }];
  
}
- (BOOL) processResponse:(WSResponse *)response {
  NSLog(@"process response");
  if (response.isFin && response.bytesLeft <= 0) {
    if (response.code == Ping) {
      [self dequeueWrite:response.buffer code:Pong writeCompletion:^{
        
      }];
    } else if (response.code == TextFrame) {
      NSString *str = [[NSString alloc] initWithData:response.buffer encoding:NSUTF8StringEncoding];
      if (str == nil) {
        [self writeError:Encoding];
        return NO;
      }
      if (self.canDispatch) {
        __weak typeof(self) weakSelf = self;
        dispatch_async(_queue, ^{
          if (weakSelf.onText) {
            weakSelf.onText(str);
          }
          NSLog(@"str:%@",str);
          [weakSelf.deleate websocketDidReceiveMessage:weakSelf text:str];
        });
      }
    } else if (response.code == BinaryFrame) {
      if (self.canDispatch) {
        NSData *data = response.buffer;
        __weak typeof(self) weakSelf = self;
        dispatch_async(_queue, ^{
          if (weakSelf.onData) {
            weakSelf.onData(data);
          }
          
          [weakSelf.deleate websocketDidReceiveData:weakSelf data:data];
        });
      }
    }
    [self.readStack removeLastObject];
  }
  return NO;
}

+ (UInt16) readUint16:(NSData *)buffer offset:(int)offset {
  Byte *byte = (Byte *)buffer.bytes;
  return (byte[offset + 0] << 8) | byte[offset + 1];
}

+ (UInt64) readUint64:(NSData *) buffer offset:(int)offset {
  Byte *byte = (Byte *)buffer.bytes;
  UInt64 value = 0;
  for (int i = 0 ; i < 7; i ++) {
    value = (value << 8) | byte[offset + i];
  }
  return value;
}

+ (void) writeUint16:(NSMutableData *)buffer offset:(int)offset value:(UInt16)value {
  Byte *byte = buffer.mutableBytes;
  byte[offset + 0] = (UInt8)((value >> 8)&0x00ff);
  byte[offset + 1] = (UInt8)((value & 0xff)&0x00ff);
}

+ (void) writeUint64:(NSMutableData *)buffer offset:(int)offset value:(UInt16)value {
  Byte *byte = buffer.mutableBytes;
  for (int i = 0; i < 8; i++) {
    byte[offset + i] = (UInt8)((value >> (8*(UInt64)(7 - i))) & 0xff);
  }
}

- (NSError *) errorWithDetail:(NSString *)detail code:(NSInteger)code {
  NSDictionary *details = @{NSLocalizedDescriptionKey:detail?:@""};
  return  [NSError errorWithDomain:ErrorDomain code:code userInfo:details];
}

#pragma mark stream delegate
- (void) stream:(NSStream *)astream handleEvent:(NSStreamEvent)eventCode {
  if (self.security && !self.certValidated && (eventCode == NSStreamEventHasBytesAvailable || eventCode == NSStreamEventHasSpaceAvailable)) {
    id possibleTrust =  [astream propertyForKey:(__bridge NSString *)kCFStreamPropertySSLPeerTrust];
    if (possibleTrust) {
      NSString *domain = [astream propertyForKey:(__bridge NSString *)kCFStreamSSLPeerName];
      if ([self.security isValid:(__bridge SecTrustRef)possibleTrust domain:domain]) {
        _certValidated = YES;
      } else {
        [self disconnectStream:[self errorWithDetail:@"Invalid SSL certificate" code:1]];
        return;
      }
    }
  }
  if (eventCode == NSStreamEventHasBytesAvailable) {
    if (astream == self.inputStream) {
      //      NSLog(@"---------");
      [self processInputStream];
    }
  }
  if (eventCode == NSStreamEventErrorOccurred) {
    [self disconnectStream:astream.streamError];
  }
  if (eventCode == NSStreamEventEndEncountered) {
    [self disconnectStream:nil];
  }
}

- (void) dealloc {
    [self.mutex lock];
    self.readyToWrite = NO;
    [self cleanupStream];
    [self.mutex unlock];
    [self.writeQueue cancelAllOperations];
}

@end


@implementation SSLCert

- (id) initWithKey:(SecKeyRef)key {
  if (self = [super init]) {
    _key = key;
  }
  return self;
}


- (id) initWithData:(NSData *)data {
  if (self = [super init]) {
    _data = data;
  }
  return self;
}

- (void) dealloc {
  CFRelease(_key);
}

@end


@implementation WSResponse



@end

@implementation SSLSecurity

- (id) initWithUsePublicKey:(BOOL)usePublicKey {
  
  NSArray *paths = [[NSBundle mainBundle] pathsForResourcesOfType:@"cer" inDirectory:@"."];
  NSMutableArray *certs = [NSMutableArray arrayWithCapacity:0];
  [paths enumerateObjectsUsingBlock:^(NSString *  _Nonnull obj, NSUInteger idx, BOOL * _Nonnull stop) {
    NSData *data = [NSData dataWithContentsOfFile:obj];
    if (!data) {
      SSLCert *sslCert = [[SSLCert alloc] initWithData:data];
      [certs addObject:sslCert];
    }
  }];
  return [[SSLSecurity alloc] initWithCerts:certs usePublicKey:usePublicKey];
}

- (id) initWithCerts:(NSArray *)certs usePublicKey:(BOOL)usePublicKey {
  if (self = [super init]) {
    _usePublicKeys = usePublicKey;
    if (self.usePublicKeys) {
      dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT,0), ^{
        NSMutableArray *publicKeys = [NSMutableArray array];
        [certs enumerateObjectsUsingBlock:^(SSLCert*  _Nonnull obj, NSUInteger idx, BOOL * _Nonnull stop) {
          if (obj.key == nil) {
            obj.key = [self extractPublicKey:obj.data];
          }
          if (obj.key) {
            [publicKeys addObject:(__bridge NSObject*)obj.key];
          }
        }];
        _pubkeys = publicKeys;
        _isRead = YES;
      });
    } else {
      NSMutableArray *certificates = [NSMutableArray array];
      [certs enumerateObjectsUsingBlock:^(SSLCert *  _Nonnull obj, NSUInteger idx, BOOL * _Nonnull stop) {
        if (obj.data) {
          [certificates addObject:obj];
        }
      }];
      _certificates = certificates;
      _isRead = YES;
    }
  }
  return self;
}

- (BOOL) isValid:(SecTrustRef )trust domain:(NSString *)domain {
  NSInteger tries = 0;
  while (!self.isRead) {
    usleep(1000);
    tries += 1;
    if (tries > 5) {
      return NO;
    }
  }
  SecPolicyRef policy;
  if (self.validateDN) {
    policy = SecPolicyCreateSSL(true, (__bridge CFStringRef)domain);
  } else {
    policy = SecPolicyCreateBasicX509();
  }
  SecTrustSetPolicies(trust,policy);
  if (self.usePublicKeys) {
    NSArray *serverPubKeys = [self publicKeyChainForTrust:trust];
    for (NSString *serverKey in serverPubKeys) {
      for (NSString *key in self.pubkeys) {
        if ([serverKey isEqualToString:key]) {
          return YES;
        }
      }
    }
  } else if (self.certificates) {
    NSArray *serverCerts = [self certificateChainForTrust:trust];
    NSMutableArray *collect = [NSMutableArray array];
    for (NSData *cert in self.certificates) {
      [collect addObject:(__bridge_transfer NSObject*)SecCertificateCreateWithData(nil, (__bridge_retained CFDataRef)cert)];
    }
    SecTrustSetAnchorCertificates(trust, (__bridge_retained CFArrayRef)collect);
    SecTrustResultType result = 0;
    SecTrustEvaluate(trust, &result);
    if ( result == kSecTrustResultUnspecified || result== kSecTrustResultProceed) {
      NSInteger trustedCount = 0;
      for (NSData *serverCert in serverCerts) {
        for (NSData *cert in self.certificates) {
          if (cert == serverCert) {
            trustedCount += 1;
            break;
          }
        }
      }
      if (trustedCount == serverCerts.count) {
        return YES;
      }
    }
  }
  return NO;
}

- (SecKeyRef )extractPublicKey:(NSData *)data {
  SecCertificateRef cert =  SecCertificateCreateWithData(nil, (__bridge CFDataRef)data);
  return [self extractPublicKeyFromCert:cert policy:SecPolicyCreateBasicX509()];
  
}

- (NSArray *) publicKeyChainForTrust:(SecTrustRef)trust {
  SecPolicyRef policy = SecPolicyCreateBasicX509();
  CFIndex index = SecTrustGetCertificateCount(trust);
  NSMutableArray *keys = [NSMutableArray array];
  for (int i = 0; i< index; i++) {
    SecCertificateRef cert = SecTrustGetCertificateAtIndex(trust, i);
    SecKeyRef key = [self extractPublicKeyFromCert:cert policy:policy];
    [keys addObject:(__bridge_transfer NSObject*)key];
  }
  return keys;
}

- (SecKeyRef)extractPublicKeyFromCert:(SecCertificateRef )cert policy:(SecPolicyRef)policy {
  SecTrustRef possibleTrust;
  SecTrustCreateWithCertificates(cert, policy,&possibleTrust);
  if (possibleTrust == nil) {
    return nil;
  }
  SecTrustResultType result = 0;
  SecTrustEvaluate(possibleTrust, &result);
  return SecTrustCopyPublicKey(possibleTrust);
}

- (NSArray *)certificateChainForTrust:(SecTrustRef)trust {
  CFIndex index = SecTrustGetCertificateCount(trust);
  NSMutableArray *certificates = [NSMutableArray array];
  for (int i = 0; i < index; i++) {
    SecCertificateRef cert = SecTrustGetCertificateAtIndex(trust, i);
    [certificates addObject:(__bridge_transfer NSData *)SecCertificateCopyData(cert)];
  }
  return certificates;
}


- (void) dealloc {
}

@end